package com.speily.study.provider.rocketmq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @Auther: SPL
 * @Date: 2019-07-10 13:23
 * @Description: rocketmq消息生产者--事物消息
 */
public class TransactionProducer {

    private final Logger logger = LoggerFactory.getLogger(TransactionProducer.class);

    private TransactionMQProducer producer;
    private String producerGroup;
    private String namesrvAddr;

    /**
     * Spring bean init-method 初始化
     */
    public void init() throws MQClientException {

        logger.info("TransactionProducer initialize!");
        logger.info(producerGroup);
        logger.info(namesrvAddr);

        // 1、初始化
        producer = new TransactionMQProducer(producerGroup);
        //2、属性设置
        producer.setNamesrvAddr(namesrvAddr);
        producer.setInstanceName(String.valueOf(System
                .currentTimeMillis()));

        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        TransactionListener transactionListener = new TransactionListenerImpl();

        producer.setExecutorService(executorService);//执行器
        producer.setTransactionListener(transactionListener);//事物监听器


        //3、启动
        producer.start();
        logger.info("TransactionProducer start success!");
    }

    /**
     * Spring bean destroy-method
     */
    public void destroy() {
        producer.shutdown();
    }

    public TransactionMQProducer getProducer() {
        return producer;
    }

    // ---------------setter -----------------

    public void setProducerGroup(String producerGroup) {
        this.producerGroup = producerGroup;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }
}
